/**
 * File: FastSequenceBoxSortedWriter.java
 * Created by: mhaimel
 * Created on: Aug 1, 2011
 * CVS:  $Id: FastSequenceBoxSortedWriter.java 1.0 Aug 1, 2011 2:23:03 PM mhaimel Exp $
 */
package uk.ac.ebi.curtain.storage.sequence.fast;

import java.io.BufferedWriter;
import java.io.Closeable;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import uk.ac.ebi.curtain.storage.CompressorStrategy;
import uk.ac.ebi.curtain.storage.sequence.OutputStreamProvider;
import uk.ac.ebi.curtain.storage.sequence.SequenceBox;
import uk.ac.ebi.curtain.storage.sequence.StreamProvider;
import uk.ac.ebi.curtain.utils.io.impl.FileIO;

/**
 * @author mhaimel
 *
 */
public class FastSequenceBoxSortedWriter implements Closeable {
	private static final char SEPARATOR = '\t';
	public static final int BUFFER_SIZE = 1024*1024*4; // 4MB
	
	private Log log = null;
	private final int bufferSize;
	private final StreamProvider<OutputStream> outStreamP;
	private final StreamProvider<OutputStream> idxStreamP;

	private final ReentrantLock lock = new ReentrantLock();
	private final AtomicReference<FastEntryWriter> writer = new AtomicReference<FastEntryWriter>();
	private final AtomicReference<DataOutputStream> outIdx = new AtomicReference<DataOutputStream>();
	private final ConcurrentSkipListMap<Long, Long> map = new ConcurrentSkipListMap<Long, Long>();
	private final AtomicLong prevIdx = new AtomicLong(0);
	private CompressorStrategy strategy = null;
	
	
	public FastSequenceBoxSortedWriter(File output){
		this(output,BUFFER_SIZE);
	}
	public FastSequenceBoxSortedWriter(final File file, int bufferSize){
		this( new OutputStreamProvider(file),new OutputStreamProvider(file.getAbsolutePath()+".qbi"),bufferSize);
	}
	
	public FastSequenceBoxSortedWriter(StreamProvider<OutputStream> outStream,StreamProvider<OutputStream> idxStream, int bufferSize) {
		this.outStreamP = outStream;
		this.idxStreamP = idxStream;
		this.bufferSize = bufferSize;
		this.strategy = CountingWriter.getDefaultStrategy();
	}
	
	private PrintWriter _doBuildStream() throws IOException {
		CompressorStrategy stg = getStrategy();
		return new PrintWriter(new BufferedWriter(new CountingWriter(
				getOutputProvider().openAndGetStream(), 
				map,stg), 
				getBufferSize()));
	}
	
	public void setStrategy(CompressorStrategy strategy) {
		this.strategy = strategy;
	}
	public CompressorStrategy getStrategy() {
		return strategy;
	}
	private int getBufferSize() {
		return this.bufferSize ;
	}
	
	public void write(Long id,SequenceBox entry) throws IOException{
		doWrite(id, entry);
	}

	private final AtomicLong prevId = new AtomicLong(-1);
	protected void doWrite(Long id, SequenceBox entry) throws IOException{
		lock.lock();
		try{
			FastEntryWriter out = _getWriter();
			long prev = prevId.getAndSet(id);
			if(prev != id){
				_changeOfId(prev,out);
			}
			out.write(id, entry);
		}finally{
			if(lock.isHeldByCurrentThread()){
				lock.unlock();
			}
		}
	}
	
	private FastEntryWriter _getWriter() throws IOException{
		return writer.get();
	}
	
	public void open()throws IOException {
		lock.lock();
		try{
			FastEntryWriter o = _getWriter();
			if(o != null){
				throw new IOException("Stream already open: " +getOutputProvider().getName());
			}
			o = new FastEntryWriter(_doBuildStream(),getSeparator());
			this.writer.set(o);
			_initIdxStream();
		}finally{
			if(lock.isHeldByCurrentThread()){
				lock.unlock();
			}
		}
	}
	
	protected void closeOutputStream() throws IOException{
		lock.lock();
		try{
			getLog().debug("Close writer " + getOutputProvider().getName());
			if(null == writer.get() ){
				throw new IOException("Stream not open for "  + outStreamP.getName());
			}
			_flush();
			FastEntryWriter o = writer.getAndSet(null);
			_closeIndexStream();
			o.getWriter().close();
		}finally{
			if(lock.isHeldByCurrentThread()){
				lock.unlock();
			}
		}
	}
	private void _closeIndexStream() throws IOException {
		DataOutputStream oi = this.outIdx.getAndSet(null);
		oi.writeLong(-1);
		FileIO.closeQuietly(oi);
	}
	
	private void _initIdxStream() throws IOException {
		this.outIdx.set(new DataOutputStream(getIndexProvider().openAndGetStream()));
		this.outIdx.get().writeInt(this.getBufferSize()); // Buffer size
		this.outIdx.get().writeLong(0l);// Start offset Unbuffered
		this.outIdx.get().writeLong(0l);// Start offset Buffered TODO
	}
	private StreamProvider<OutputStream> getIndexProvider() {
		return this.idxStreamP;
	}	
	
	public static char getSeparator(){
		return SEPARATOR;
	}
	
	private void _flush() throws IOException{
		_changeOfId(this.prevId.getAndSet(-1),_getWriter());
	}
	
	private void _changeOfId(long prev, FastEntryWriter out) throws IOException {
		if(prev == -1){
			return;
		}
		out.getWriter().flush();
		DataOutputStream oi = this.outIdx.get();
		oi.writeLong(prev);
		Long lastIdx = this.map.keySet().last();
		long prevIdx = this.prevIdx.getAndSet(lastIdx);
		
		oi.writeLong(lastIdx);
		
		ConcurrentNavigableMap<Long, Long> tailMap = this.map.tailMap(prevIdx, false);
		for(Entry<Long, Long> s : tailMap.entrySet()){
			oi.writeLong(s.getValue());
		}
		this.map.clear();
	}
	
	public Log getLog() {
		if(null == log){
			log = LogFactory.getLog(this.getClass());
		}
		return log;
	}
	
	private static class CountingWriter extends FastBlockWriter{
		private final Map<Long, Long> map;
		private final CompressorStrategy strategy;

		public CountingWriter(OutputStream o, Map<Long, Long> entry,CompressorStrategy strategy) throws FileNotFoundException {
			super(o, strategy);
			this.map = entry;
			this.strategy = strategy;
		}
		
		@Override
		protected CompressorStrategy getStrategy() {
			if(null == this.strategy){
				return super.getStrategy();
			}
			return this.strategy;
		}

		@Override
		protected void writeEvent(long uncomp, long comp) {
			map.put(uncomp, comp);
			super.writeEvent(uncomp, comp);
		}
	}

	@Override
	public void close() throws IOException {
		this.closeOutputStream();
	}
	private StreamProvider<OutputStream> getOutputProvider() {
		return this.outStreamP;
	}

}
